package com.hopu.user.era;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.era.TotalCartBean;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.IOException;


public class TotalCart {

    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("totalCart").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");
        JavaPairRDD<TotalCartBean, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String era = json.getJSONObject("birthday").getString("year").substring(2, 3) + "0";
            String event_type = json.getString("event_type");
            return new Tuple2<>(new TotalCartBean(era, event_type, 1), 1);
        });

        //过滤出各年代的cart类型的操作
        String[] eras = {"50", "60", "70", "80", "90", "00", "10"};
        for (String era: eras) {
            transAndPut(pairRDD, era);
        }
    }

    public static void transAndPut(JavaPairRDD<TotalCartBean, Integer> pairRDD, String era) {
        //过滤出某个年代的cart操作
        pairRDD = pairRDD.filter(t -> t._1.getEra().equals(era) && t._1.getEvent_type().equals("cart"));
        JavaPairRDD<TotalCartBean, Iterable<Integer>> group = pairRDD.groupByKey();

        JavaRDD<TotalCartBean> map = group.map(t -> {
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            t._1.setCount(count);
            return t._1;
        });

        TotalCartBean first = map.first();
        System.out.println(first);

        //插入数据到HBase中
        String[] columns = {"era", "event_type", "count"};
        String[] values = {first.getEra(), first.getEvent_type(), first.getCount()+""};

        try {
            HBaseUtils.putOneRowToHbase("era", first.getEra(), "totalCart", columns, values);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
